#ifndef _H_CARMELA_BROKER_
#define _H_CARMELA_BROKER_

#ifdef __cplusplus
extern "C" {
#endif

#include "carmela_util.h"
#include "carmela_sdk.h"

struct MQBroker ;
struct MQStore ;
struct MQTopic ;
struct MQSubscribeGroup ;
struct MQFileMapping ;

/*
 * file restore
 */

/*
$HOME/carmela_data/
	{store_name}/
		{msg_offset(%020PRIu64):00000000000000000000}.cmqm
		{delay_msg_offset(%020PRIu64):00000000000000000000}.cmqd
		{topic{/
			{topic_offset(%020PRIu64):00000000000000000000}.cmqt
*/

#define MQ_MSG_FILE_RESTORE_FORMAT_VERSION_1		'1'
#define MQ_TOPIC_FILE_RESTORE_FORMAT_VERSION_1		'1'
#define MQ_DELAY_MSG_FILE_RESTORE_FORMAT_VERSION_1	'1'

#define MQ_MSG_FILE_HEADER_FORMAT_MAGIC_OFFSET			0
#define MQ_MSG_FILE_HEADER_FORMAT_MAGIC_LENGTH			4
#define MQ_MSG_FILE_HEADER_FORMAT_VERSION_OFFSET		(MQ_MSG_FILE_HEADER_FORMAT_MAGIC_OFFSET+MQ_MSG_FILE_HEADER_FORMAT_MAGIC_LENGTH)
#define MQ_MSG_FILE_HEADER_FORMAT_VERSION_LENGTH		1
#define MQ_MSG_FILE_HEADER_FORMAT_MSG_OFFSET_OFFSET		(MQ_MSG_FILE_HEADER_FORMAT_VERSION_OFFSET+MQ_MSG_FILE_HEADER_FORMAT_VERSION_LENGTH)
#define MQ_MSG_FILE_HEADER_FORMAT_MSG_OFFSET_LENGTH		8
#define MQ_MSG_FILE_HEADER_FORMAT_ORI_FILE_SIZE_OFFSET		(MQ_MSG_FILE_HEADER_FORMAT_MSG_OFFSET_OFFSET+MQ_MSG_FILE_HEADER_FORMAT_MSG_OFFSET_LENGTH)
#define MQ_MSG_FILE_HEADER_FORMAT_ORI_FILE_SIZE_LENGTH		8
#define MQ_MSG_FILE_HEADER_FORMAT_IS_CLEAN_OFFSET		(MQ_MSG_FILE_HEADER_FORMAT_ORI_FILE_SIZE_OFFSET+MQ_MSG_FILE_HEADER_FORMAT_ORI_FILE_SIZE_LENGTH)
#define MQ_MSG_FILE_HEADER_FORMAT_IS_CLEAN_LENGTH		1

#define MQ_MSG_FILE_HEADER_FORMAT_MAGIC_VALUE			"CMQM"

#define MQ_MSG_FILE_HEADER_FORMAT_NOT_CLEAN			'0'
#define MQ_MSG_FILE_HEADER_FORMAT_CLEANNED			'1'

/*
msg head format :
			msg_file_magic		string		4 bytes		'CMQM'
			msg_file_version	uint8_t		1 byte		'0'
			msg_file_offset		uint64_t	8 bytes
			msg_file_size		uint64_t	8 bytes
			is_clean		string		1 byte		'0':no '1':yes
			reserved		string		10 byte
								32

msg unit format :
			msg_offset		uint64_t	8 bytes
			topic_len		uint8_t		1 byte
			topic			string		topic_len bytes
			tags_len		uint16_t	2 bytes
			tags			string		tags_len bytes
			msg_data_len		uint32_t	4 bytes
			msg_data		string		msg_data_len bytes
								15+topic_len+tags_len+msg_data_len - demo 37
*/

#define MQ_TOPIC_FILE_HEADER_FORMAT_MAGIC_OFFSET		0
#define MQ_TOPIC_FILE_HEADER_FORMAT_MAGIC_LENGTH		4
#define MQ_TOPIC_FILE_HEADER_FORMAT_VERSION_OFFSET		(MQ_TOPIC_FILE_HEADER_FORMAT_MAGIC_OFFSET+MQ_TOPIC_FILE_HEADER_FORMAT_MAGIC_LENGTH)
#define MQ_TOPIC_FILE_HEADER_FORMAT_VERSION_LENGTH		1
#define MQ_TOPIC_FILE_HEADER_FORMAT_MSG_OFFSET_OFFSET		(MQ_TOPIC_FILE_HEADER_FORMAT_VERSION_OFFSET+MQ_TOPIC_FILE_HEADER_FORMAT_VERSION_LENGTH)
#define MQ_TOPIC_FILE_HEADER_FORMAT_MSG_OFFSET_LENGTH		8
#define MQ_TOPIC_FILE_HEADER_FORMAT_ORI_FILE_SIZE_OFFSET	(MQ_TOPIC_FILE_HEADER_FORMAT_MSG_OFFSET_OFFSET+MQ_TOPIC_FILE_HEADER_FORMAT_MSG_OFFSET_LENGTH)
#define MQ_TOPIC_FILE_HEADER_FORMAT_ORI_FILE_SIZE_LENGTH	8
#define MQ_TOPIC_FILE_HEADER_FORMAT_IS_CLEAN_OFFSET		(MQ_TOPIC_FILE_HEADER_FORMAT_ORI_FILE_SIZE_OFFSET+MQ_TOPIC_FILE_HEADER_FORMAT_ORI_FILE_SIZE_LENGTH)
#define MQ_TOPIC_FILE_HEADER_FORMAT_IS_CLEAN_LENGTH		1
#define MQ_TOPIC_FILE_HEADER_FORMAT_RESERVED_OFFSET		(MQ_TOPIC_FILE_HEADER_FORMAT_IS_CLEAN_OFFSET+MQ_TOPIC_FILE_HEADER_FORMAT_IS_CLEAN_LENGTH)
#define MQ_TOPIC_FILE_HEADER_FORMAT_RESERVED_LENGTH		10

#define MQ_TOPIC_FILE_HEADER_FORMAT_MAGIC_VALUE			"CMQT"

#define MQ_TOPIC_FILE_HEADER_FORMAT_NOT_CLEAN			'0'
#define MQ_TOPIC_FILE_HEADER_FORMAT_CLEANNED			'1'

/*
topic head format :
			topic_magic		string		4 bytes		'CMQT'
			topic_file_version	uint8_t		1 byte		'0'
			topic_file_offset	uint64_t	8 bytes
			topic_file_size		uint64_t	8 bytes
			is_clean		string		1 byte		'0':no '1':yes
			reserved		string		10 byte
								32

topic unit format :
			topic_offset		uint64_t	8 bytes
			msg_offset		uint64_t	8 bytes
								8 - demo 8
*/

#define MQ_DELAY_MSG_FILE_HEADER_FORMAT_MAGIC_OFFSET		0
#define MQ_DELAY_MSG_FILE_HEADER_FORMAT_MAGIC_LENGTH		4
#define MQ_DELAY_MSG_FILE_HEADER_FORMAT_VERSION_OFFSET		(MQ_DELAY_MSG_FILE_HEADER_FORMAT_MAGIC_OFFSET+MQ_DELAY_MSG_FILE_HEADER_FORMAT_MAGIC_LENGTH)
#define MQ_DELAY_MSG_FILE_HEADER_FORMAT_VERSION_LENGTH		1
#define MQ_DELAY_MSG_FILE_HEADER_FORMAT_MSG_OFFSET_OFFSET	(MQ_DELAY_MSG_FILE_HEADER_FORMAT_VERSION_OFFSET+MQ_DELAY_MSG_FILE_HEADER_FORMAT_VERSION_LENGTH)
#define MQ_DELAY_MSG_FILE_HEADER_FORMAT_MSG_OFFSET_LENGTH	8
#define MQ_DELAY_MSG_FILE_HEADER_FORMAT_ORI_FILE_SIZE_OFFSET	(MQ_DELAY_MSG_FILE_HEADER_FORMAT_MSG_OFFSET_OFFSET+MQ_DELAY_MSG_FILE_HEADER_FORMAT_MSG_OFFSET_LENGTH)
#define MQ_DELAY_MSG_FILE_HEADER_FORMAT_ORI_FILE_SIZE_LENGTH	8
#define MQ_DELAY_MSG_FILE_HEADER_FORMAT_IS_CLEAN_OFFSET		(MQ_DELAY_MSG_FILE_HEADER_FORMAT_ORI_FILE_SIZE_OFFSET+MQ_DELAY_MSG_FILE_HEADER_FORMAT_ORI_FILE_SIZE_LENGTH)
#define MQ_DELAY_MSG_FILE_HEADER_FORMAT_IS_CLEAN_LENGTH		1
#define MQ_DELAY_MSG_FILE_HEADER_FORMAT_RESERVED_OFFSET		(MQ_DELAY_MSG_FILE_HEADER_FORMAT_IS_CLEAN_OFFSET+MQ_DELAY_MSG_FILE_HEADER_FORMAT_IS_CLEAN_LENGTH)
#define MQ_DELAY_MSG_FILE_HEADER_FORMAT_RESERVED_LENGTH		10

#define MQ_DELAY_MSG_FILE_HEADER_FORMAT_MAGIC_VALUE		"CMQD"

#define MQ_DELAY_MSG_FILE_HEADER_FORMAT_NOT_CLEAN		'0'
#define MQ_DELAY_MSG_FILE_HEADER_FORMAT_CLEANNED		'1'

/*
delay msg head format :
			delay_msg_file_magic	string		4 bytes		'CMQD'
			delay_msg_file_version	uint8_t		1 byte		'0'
			delay_msg_file_offset	uint64_t	8 bytes
			delay_msg_file_size	uint64_t	8 bytes
			is_clean		string		1 byte		'0':no '1':yes
			reserved		string		10 byte
								32

delay msg unit format :
			delay_msg_offset	uint64_t	8 bytes
			delay_seconds		uint32_t	4 bytes
			is_sent			string		1 byte		'0':not send , '1':sent
			store_len		uint8_t		1 byte
			store			string		store_len bytes
			topic_len		uint8_t		1 byte
			topic			string		topic_len bytes
			tags_len		uint16_t	2 bytes
			tags			string		tags_len bytes
			msg_data_len		uint32_t	4 bytes
			msg_data		string		msg_data_len bytes
								19+topic_len+tags_len+msg_data_len - demo 41
*/

/*
 * file mapping
 */

struct MQFileMapping
{
	struct list_head	file_list_node ;
	
	char			path_file_name[ PATH_MAX ] ;
	ssize_t			mmap_size ;
	char			*file_mmap ;
	
	size_t			begin_offset ;
	size_t			file_size ;
	char			*fill_ptr ;
	char			*over_end_ptr ;
	
	struct list_head	subscribe_session_list ; /* struct MQAcceptedSession */
} ;

/*
 * store
 */

#define MSG_FILE_LIST_NODES_COUNT_SOFT_MAX_DEFAULT		10
#define MSG_FILE_LIST_NODES_COUNT_HARD_MAX_DEFAULT		20

#define DELAY_MSG_FILE_LIST_NODES_COUNT_SOFT_MAX_DEFAULT	10
#define DELAY_MSG_FILE_LIST_NODES_COUNT_HARD_MAX_DEFAULT	20

#define DELAY_MSG_FILE_LIST_NODES_COUNT_SOFT_MAX_DEFAULT	10

struct MQStore
{
	char			store_name[ MQ_STORE_NAME_MAX_LEN+1 ] ;
	uint8_t			store_name_len ;
	struct rb_node		store_name_tree_node ; /* struct MQStore */
	
	char			store_path[ PATH_MAX ] ;
	
	uint64_t		max_msg_offset ;
	
	struct list_head	msg_file_list ; /* struct MQFileMapping */
	size_t			msg_file_list_nodes_count ;
	
	struct rb_root		topic_name_tree ; /* struct MQTopic */
	
	uint64_t		max_delay_msg_offset ;
	
	struct list_head	delay_msg_file_list ; /* struct MQFileMapping */
	size_t			delay_msg_file_list_nodes_count ;
	struct MQFileMapping	*first_not_sent_delay_msg_file ;
	struct MQFileMapping	*last_not_sent_delay_msg_file ;
	struct rb_root		delay_msg_offset_tree ;
	struct rb_root		delay_msg_send_timestamp_tree ;
} ;

int LinkStoreNameTreeNode( struct MQBroker *broker , struct MQStore *store );
void UnlinkStoreNameTreeNode( struct MQBroker *broker , struct MQStore *store );
struct MQStore *QueryStoreNameWithLenTreeNode( struct MQBroker *broker , struct MQStore *store );
struct MQStore *TravelStoreNameTreeNode( struct MQBroker *broker , struct MQStore *store );
void DestroyStoreNameTree( struct MQBroker *broker );

int LoadAllMessageStores( struct MQBroker *broker );
int UnloadAllMessageStores( struct MQBroker *broker );

/*
 * message
 */

#define MQ_MSG_FILE_SIZE_DEFAULT	100*1024*1024
#define MQ_MSG_FILE_HEAD_SIZE		32

size_t CalcMessageDataUnitSize( uint8_t topic_name_len , uint16_t tags_len , uint32_t msg_data_len );
uint64_t MemWrite_MessageDataUnit( char *unit_offset_ptr , uint64_t msg_offset , uint8_t topic_name_len , char *topic_name , uint16_t tags_len , char *tags , uint32_t msg_data_len , char *msg_data );
uint64_t MemRead_MessageDataUnit( char *unit_offset_ptr , uint64_t *p_msg_offset , uint8_t *p_topic_name_len , char **pp_topic_name , uint16_t *p_tags_len , char **pp_tags , uint32_t *p_msg_data_len , char **pp_msg_data );
int CreateMessageDataFile( struct AppendBuffer *comm_data , struct MQBroker *broker , struct MQStore *store , size_t begin_offset , struct MQFileMapping **p_msg_file ,
size_t *p_file_size );
int LoadMessageDataFile( struct AppendBuffer *comm_data , struct MQBroker *broker , struct MQStore *store , size_t begin_offset , char *filename , struct MQFileMapping **p_msg_file , size_t *p_file_size );
int LoadAllMessageDataFiles( struct MQBroker *broker , struct MQStore *store );
int UnloadMessageDataFile( struct MQBroker *broker , struct MQStore *store , struct MQFileMapping *msg_file );
int CleanMessageDataFile( struct MQBroker *broker , struct MQStore *store , struct MQFileMapping *msg_file );
int UnloadAllMessageDataFiles( struct MQBroker *broker , struct MQStore *store );

/*
 * topic
 */

#define TOPIC_FILE_LIST_NODES_COUNT_SOFT_MAX_DEFAULT	10
#define TOPIC_FILE_LIST_NODES_COUNT_HARD_MAX_DEFAULT	20

#define MQ_TOPIC_FILE_SIZE_DEFAULT	10*1024*1024
#define MQ_TOPIC_FILE_HEAD_SIZE		32

struct MQTopic
{
	char			topic_name[ MQ_TOPIC_NAME_MAX_LEN+1 ] ;
	uint8_t			topic_name_len ;
	struct rb_node		topic_name_tree_node ; /* struct MQTopic */
	
	char			topic_path[ PATH_MAX ] ;
	
	uint64_t		max_topic_offset ;
	
	struct list_head	topic_file_list ; /* struct MQFileMapping */
	size_t			topic_file_list_nodes_count ;
	
	struct rb_root		subscribe_group_name_tree ; /* struct MQSubscribeGroup */
} ;

int LinkTopicNameTreeNode( struct MQStore *store , struct MQTopic *topic );
void UnlinkTopicNameTreeNode( struct MQStore *store , struct MQTopic *topic );
struct MQTopic *QueryTopicNameWithLenTreeNode( struct MQStore *store , struct MQTopic *topic );
struct MQTopic *TravelTopicNameTreeNode( struct MQStore *store , struct MQTopic *topic );
void DestroyTopicNameTree( struct MQStore *store );
size_t CalcTopicDataUnitSize();
uint64_t MemWrite_TopicDataUnit( char *unit_offset_ptr , uint64_t topic_offset , uint64_t msg_offset );
uint64_t MemRead_TopicDataUnit( char *unit_offset_ptr , uint64_t *p_topic_offset , uint64_t *p_msg_offset );
int CreateTopicDataFile( struct AppendBuffer *comm_data , struct MQBroker *broker , struct MQStore *store , struct MQTopic *topic , size_t begin_offset , struct MQFileMapping **pp_topic_file , size_t *p_file_size );
int LoadTopicDataFile( struct AppendBuffer *comm_data , struct MQBroker *broker , struct MQStore *store , struct MQTopic *topic , size_t begin_offset , char *filename , struct MQFileMapping **pp_topic_file , size_t *p_file_size );
int UnloadTopicDataFile( struct MQBroker *broker , struct MQStore *store , struct MQTopic *topic , struct MQFileMapping *topic_file );
int CleanTopicDataFile( struct MQBroker *broker , struct MQStore *store , struct MQTopic *topic , struct MQFileMapping *topic_file );
int LoadAllTopicDataFiles( struct MQBroker *broker , struct MQStore *store , struct MQTopic *topic );
int LoadAllTopics( struct MQBroker *broker , struct MQStore *store );
int UnloadAllTopics( struct MQBroker *broker , struct MQStore *store );

/*
 * delay message
 */

#define MQ_DELAY_MSG_FILE_SIZE_DEFAULT	10*1024*1024
#define MQ_DELAY_MSG_FILE_HEAD_SIZE	32

#define MQ_DELAY_MSG_WAIT_TO_SEND	'W'
#define MQ_DELAY_MSG_IS_SENT		'T'
#define MQ_DELAY_MSG_ROLLBACK		'R'
#define MQ_DELAY_MSG_CANCEL		'C'

struct MQDelayMessage
{
	unsigned char			custom_evnet_session ;
	
	uint64_t			delay_msg_offset ;
	struct rb_node			delay_msg_offset_tree_node ; /* struct MQDelayMessage */
	
	uint32_t			delay_seconds ;
	struct rb_node			delay_msg_send_timestamp_tree_node ; /* struct MQDelayMessage */
	
	int				delay_timer ;
	
	struct MQFileMapping		*delay_msg_file ;
	char				*delay_msg_file_unit_base ;
} ;

int LinkDelayMessageOffsetTreeNode( struct MQStore *store , struct MQDelayMessage *delay_msg );
void UnlinkDelayMessageOffsetTreeNode( struct MQStore *store , struct MQDelayMessage *delay_msg );
struct MQDelayMessage *QueryDelayMessageOffsetTreeNode( struct MQStore *store , struct MQDelayMessage *delay_msg );
void DestroyDelayMessageOffsetTree( struct MQStore *store );

int LinkDelayMessageSendTimestampTreeNode( struct MQStore *store , struct MQDelayMessage *delay_msg );
void UnlinkDelayMessageSendTimestampTreeNode( struct MQStore *store , struct MQDelayMessage *delay_msg );
struct MQDelayMessage *FindRecentSendTimestampDelayMessage( struct MQStore *store );
struct MQDelayMessage *TravelDelayMessageSendTimestampTreeNode( struct MQStore *store , struct MQDelayMessage *delay_msg );

struct MQDelayMessage *CreateDelayMessage( struct TcpdaemonServerEnvironment *p_env , struct AppendBuffer *comm_data , struct MQFileMapping *last_delay_msg_file , char *fill_ptr , uint64_t delay_msg_offset , uint32_t delay_seconds );
void DestroyDelayMessage( struct TcpdaemonServerEnvironment *p_env , struct MQDelayMessage *delay_msg );
void MemSet_DelayMessageDataUnit_is_sent( char *unit_offset_ptr , char is_sent );
size_t CalcDelayMessageDataUnitSize( uint8_t topic_len , uint16_t tags_len , uint32_t msg_data_len );
uint64_t MemWrite_DelayMessageDataUnit( char *unit_offset_ptr , uint64_t delay_msg_id , uint32_t delay_seconds , char is_sent , uint8_t store_name_len , char *store_name , uint8_t topic_name_len , char *topic_name , uint16_t tags_len , char *tags , uint32_t msg_data_len , char *msg_data );
uint64_t MemRead_DelayMessageDataUnit( char *unit_offset_ptr , uint64_t *p_delay_msg_id , uint32_t *p_delay_seconds , char *p_is_sent , uint8_t *p_store_name_len , char **pp_store_name , uint8_t *p_topic_name_len , char **pp_topic_name , uint16_t *p_tags_len , char **pp_tags , uint32_t *p_msg_data_len , char **pp_msg_data );
int CreateDelayMessageDataFile( struct AppendBuffer *comm_data , struct MQBroker *broker , struct MQStore *store , size_t begin_offset , struct MQFileMapping **pp_delay_msg_file , size_t *p_file_size );
int LoadDelayMessageDataFile( struct AppendBuffer *comm_data , struct MQBroker *broker , struct MQStore *store , size_t begin_offset , char *filename , struct MQFileMapping **pp_delay_msg_file , size_t *p_file_size );
int LoadAllDelayMessageDataFiles( struct MQBroker *broker , struct MQStore *store );
int UnloadDelayMessageDataFile( struct MQBroker *broker , struct MQStore *store , struct MQFileMapping *delay_msg_file );
int CleanDelayMessageDataFile( struct MQBroker *broker , struct MQStore *store , struct MQFileMapping *delay_msg_file );
int UnloadAllDelayMessageDataFiles( struct MQBroker *broker , struct MQStore *store );

#ifdef __cplusplus
}
#endif

#endif

